package com.tairanchina.md.elife.rabbit.client.producer.impl;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import com.tairanchina.md.elife.rabbit.client.producer.RabbitProducerClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.concurrent.TimeoutException;

/**
 * Created by lxliuxuan on 2016/12/12.
 */

public class TXProducerRabbitClient extends RabbitProducerClient
{
    private Logger logger = LoggerFactory.getLogger(TXProducerRabbitClient.class);

    //获取连接工厂
    public ConnectionFactory getConnectionFactory()
    {

        if (factory == null )
        {
            initConnectionFactory();
        }
        return factory;

    }

    //获取连接，连接保持长连接，不同的用户请求提供不同的chnnel
    public Connection getConnection()
    {

        if (connection == null || !connection.isOpen())
        {
            initConnection();
        }
        return connection;

    }

    public void sendMessage(String exchangeName, String queueKey, Object obj) throws Exception {
        Channel channel = null;

        try
        {
            channel = getChannel();
            channel.basicPublish(exchangeName, queueKey, MessageProperties.PERSISTENT_TEXT_PLAIN,
                                 handlerEncode(obj));
        }
        catch (Exception e)
        {
            e.printStackTrace();
            logger.error("消息发送失败。e:{}", e);
            throw new Exception(e);
        }
        finally
        {
            try
            {
                if (channel != null)
                {
                    channel.close();
                }
            }
            catch (TimeoutException | IOException e)
            {
                e.printStackTrace();
                logger.error("生产者端消息通道关闭失败。e:{}", e);
            }
        }
    }



    /**
     * 消息编码算法
     *
     * @param obj
     * @return
     */
    protected byte[] handlerEncode(Object obj) throws Exception
    {
        ByteArrayOutputStream bos = null;
        ObjectOutputStream oos = null;
        byte[] result = null;
        try
        {
            bos = new ByteArrayOutputStream();
            oos = new ObjectOutputStream(bos);
            oos.writeObject(obj);
            result = bos.toByteArray();
        }
        catch (Exception e)
        {
            e.printStackTrace();
            logger.error("消息编码失败。e:{}---data:{}", e, obj);
            throw new Exception(e);
        }
        finally
        {
            try
            {
                if (bos != null)
                {
                    bos.close();
                }
            }
            catch (IOException e1)
            {
                e1.printStackTrace();
                logger.error("数据输出流1关闭失败。e:{}", e1);
            }

            try
            {
                if (oos != null)
                {
                    oos.close();
                }
            }
            catch (IOException e2)
            {
                e2.printStackTrace();
                logger.error("数据输出流2关闭失败。e:{}", e2);
            }
        }
        return result;
    }



    //获取连接chaanel 使用Pool进行封装
    public Channel getChannel()
    {

        Channel channel = null;
        try
        {

            channel = getConnection().createChannel();
        }
        catch (IOException e)
        {
            e.printStackTrace();
            logger.error("获取通道失败。e:{}", e);
        }
        return channel;
    }

    //初始化RabbitMQ
    public void init()
    {
        initConnectionFactory();
        initConnection();
    }

    public void initConnection()
    {
        try
        {
            connection = getConnectionFactory().newConnection();
        }
        catch (Exception e)
        {
            e.printStackTrace();
            logger.error("RabbitMQ 连接初始化失败 e:{}", e);
        }
    }

    public void initConnectionFactory()
    {
        factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setUsername(username);
        factory.setPassword(password);
        factory.setVirtualHost(virtualhost);
        factory.setPort(port);
        //设置自动重连
        factory.setAutomaticRecoveryEnabled(true);
    }

}
